package com.atguigu.sparkstreaming.demos

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
 * Created by Smexy on 2022/8/22
 *

    进行一个有状态的计算:  累加。
        指定状态存储的位置:    存储在redis中

 *    把单词统计的结果输出到redis中。
 *
 *
 */
object ForeachRDDDemo {

  def main(args: Array[String]): Unit = {


    val streamingContext = new StreamingContext("local[*]", "wordcount", Seconds(5))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "sz220409test",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicD")


    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val ds1: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map(word => (word, 1))
      .reduceByKey(_ + _)

    ds1.cache()

   /*
      输出到redis，并且和之前批次已经输出的状态进行累加

      设计redis中的key-value: 参考粒度

          所有的单词是一个K-V：
                key:  wordcount
                value:   hash(map)

         一个单词一个K-V：
                key:   wc:单词
                value:  个数

         -----------------
         只要是获取数据库连接，以分区为单位获取。 一个分区共享一个连接，而不是分区中的每个数据各自用各自的。
    */

    //直接写出到redis
    ds1.foreachRDD(rdd => {

      rdd.foreachPartition(partition => {

        //一个分区创建一个连接
        val jedis = new Jedis("hadoop102", 6379)

        /*
            使用连接

            如果遍历的集合中的元素是 ()tuple，都可以用模式匹配来遍历。
                DStream[(String, Int)].foreach( (x:(String, Int) )  => {
                    val key = x._1
                    val value =x._2
                })

                等价于
                DStream[(String, Int)].foreach(
                  case (word, count) => {
                      val key = word
                      val value = count
                  }
                })
         */
        partition.foreach{
              //在当前word的基础上，再加上count，就是一个有状态的计算。借助redis提供的api实现了状态的累加
          case (word, count) => jedis.hincrBy("wordcount",word,count)
        }


        //使用完了
        jedis.close()

      })

    })

    //如果还希望在屏幕看到打印
    ds1.print(1000)

    //⑥启动App
    streamingContext.start()

    //等待发送终止命令，不发送，阻塞在这行，一直等
    streamingContext.awaitTermination()

  }

}
